package com.vxdata.utils.modules.hbase.util;

import cn.hutool.db.PageResult;
import com.alibaba.fastjson2.JSON;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.*;

/**
 * @author lijj
 * @date 2019-08-27
 */
public class HBaseUtil {

    /**
     * 创建HBase表.
     *
     * @param tableName 表名
     * @param cfs       列族的数组
     * @return 是否创建成功
     */
    public static boolean createTable(String tableName, List<String> cfs) {
        try (HBaseAdmin admin = (HBaseAdmin) HBaseConn.getHBaseConn().getAdmin()) {//获取操作对象admin
            if (admin.tableExists(TableName.valueOf(tableName))) {
                return false;
            }
            List<ColumnFamilyDescriptor> familyDescriptors = new ArrayList<>(cfs.size());
            for (String column : cfs) {
                familyDescriptors.add(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(column)).build());
            }
            TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName))
                    .setColumnFamilies(familyDescriptors).build();
            admin.createTable(tableDescriptor);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }

    /**
     * 查看HBase表是否存在
     *
     * @param tableName 表名
     * @return 是否存在该表
     */
    public static boolean isTableExist(String tableName) {
        try (HBaseAdmin admin = (HBaseAdmin) HBaseConn.getHBaseConn().getAdmin()) {
            if (admin.tableExists(TableName.valueOf(tableName))) {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }


    /**
     * 删除hbase表.
     *
     * @param tableName 表名
     * @return 是否删除成功
     */
    public static boolean deleteTable(String tableName) {
        try (HBaseAdmin admin = (HBaseAdmin) HBaseConn.getHBaseConn().getAdmin()) {
            admin.disableTable(TableName.valueOf(tableName));
            admin.deleteTable(TableName.valueOf(tableName));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }

    /**
     * hbase插入一条数据.
     * put参数 family列族名  column列名本身 val/data 列值
     * String family="info" string column= "name" string val = "zhangsan"
     *
     * @param tableName 表名
     * @param rowKey    行名
     * @param cfName    列族名（family)
     * @param qualifier 列名（数组）
     * @param data      数据 列值
     * @return 是否插入成功
     */
    public static boolean putRow(String tableName, String rowKey, String cfName, String qualifier,
                                 String data) {
        try (Table table = HBaseConn.getTable(tableName)) {
            Put put = new Put(Bytes.toBytes(rowKey));//把字符串变成字节序列
            put.addColumn(Bytes.toBytes(cfName), Bytes.toBytes(qualifier), Bytes.toBytes(data));//增加列
            table.put(put);
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }
        return true;
    }

    public static boolean putRows(String tableName, List<Put> puts) {
        try (Table table = HBaseConn.getTable(tableName)) {
            table.put(puts);
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }
        return true;
    }

    public static List<Map<String, Object>> getAllData(String tablename) throws IOException {
        List<Map<String, Object>> records = new ArrayList<>();

        TableName name = TableName.valueOf(tablename);
        Table table = HBaseConn.getHBaseConn().getTable(name);
        ResultScanner scanner = table.getScanner(new Scan());
        for (Result result : scanner) {
            Map<String, Object> map = new LinkedHashMap<>();

            for (Cell cell : result.rawCells()) {
                // 获取行键、列族、列名和值
                String row = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
                String family = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
                String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                map.put("row", row);
                map.put(family + ":" + qualifier, value);
            }
            // 将 Map 添加到 List 中
            records.add(map);
        }
        return records;


    }

    /**
     * 查找一行记录
     *
     * @param tablename 表名
     * @param rowKey    行名
     */
    public static String selectRow(String tablename, String rowKey) throws IOException {
        String record = "";
        TableName name = TableName.valueOf(tablename);
        Table table = HBaseConn.getHBaseConn().getTable(name);
        Get g = new Get(rowKey.getBytes());
        Result rs = table.get(g);
        NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> map = rs.getMap();
        for (Cell cell : rs.rawCells()) {
            StringBuffer stringBuffer = new StringBuffer().append(Bytes.toString(cell.getRowArray())).append("\t")
                    .append(Bytes.toString(cell.getFamilyArray())).append("\t")
                    .append(Bytes.toString(cell.getQualifierArray())).append("\t")
                    .append(Bytes.toString(cell.getValueArray())).append("\n");
            String str = stringBuffer.toString();
            record += str;
        }
        return record;
    }

    /**
     * 查找单行单列族单列记录
     *
     * @param tablename    表名
     * @param rowKey       行名
     * @param columnFamily 列族名
     * @param column       列名
     * @return
     */
    public static String selectValue(String tablename, String rowKey, String columnFamily, String column) throws IOException {
        TableName name = TableName.valueOf(tablename);
        Table table = HBaseConn.getHBaseConn().getTable(name);
        Get g = new Get(rowKey.getBytes());
        g.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));
        Result rs = table.get(g);
        return Bytes.toString(rs.value());
    }

    /**
     * 分页查询
     *
     * @param tableName    表名
     * @param pageNum      第几页
     * @param pageSize     每页条数
     * @param columnFamily 列族名
     * @param columnName   列名
     * @return
     * @throws IOException
     */
    public static List<Map<String, String>> queryPage(String tableName, int pageNum, int pageSize, String columnFamily, String columnName) throws IOException {
        int maxSize = pageNum * pageSize;
        PageResult pageResult = queryPage(tableName, null, maxSize, columnFamily, columnName);
        List<Map<String, String>> resultsMapList = pageResult.getResultsAsMaps();
        int totalSize = resultsMapList.size();
        int fromIndex = Math.max(totalSize - pageSize, 0);
        return resultsMapList.subList(fromIndex, totalSize);
    }


    /**
     * 分页查询 HBase 表的数据
     *
     * @param tableName    表名
     * @param startRow     起始行键
     * @param pageSize     每页的大小
     * @param columnFamily 列族名（可选）
     * @param columnName   列名（可选）
     * @return 当前页的数据和下一页的起始行键
     * @throws IOException
     */
    public static PageResult queryPage(String tableName, String startRow, int pageSize, String columnFamily, String columnName) throws IOException {
        List<Result> results = new ArrayList<>();
        Table table = HBaseConn.getHBaseConn().getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();

        if (startRow != null && !startRow.isEmpty()) {
            scan.withStartRow(Bytes.toBytes(startRow));
        }

        // 设置分页过滤器
        PageFilter pageFilter = new PageFilter(pageSize + 1); // 多取一条，用于判断是否有下一页
        scan.setFilter(pageFilter);

        // 如果指定了列族名和列名，则只扫描这些列
        if (columnFamily != null && !columnFamily.isEmpty()) {
            if (columnName != null && !columnName.isEmpty()) {
                scan.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName));
            } else {
                scan.addFamily(Bytes.toBytes(columnFamily));
            }
        }

        ResultScanner scanner = table.getScanner(scan);
        Iterator<Result> iterator = scanner.iterator();

        String nextStartRow = null;
        int count = 0;

        while (iterator.hasNext()) {
            Result result = iterator.next();
            if (count < pageSize) {
                results.add(result);
            } else {
                nextStartRow = Bytes.toString(result.getRow());
                break;
            }
            count++;
        }
        return new PageResult(results, nextStartRow);
    }

    /**
     * 封装分页结果
     */
    public static class PageResult {
        private List<Result> results;
        private String nextStartRow;

        public PageResult(List<Result> results, String nextStartRow) {
            this.results = results;
            this.nextStartRow = nextStartRow;
        }

        public List<Result> getResults() {
            return results;
        }

        public String getNextStartRow() {
            return nextStartRow;
        }

        public List<Map<String, String>> getResultsAsMaps() {
            List<Map<String, String>> serializedResults = new ArrayList<>();
            for (Result result : results) {
                Map<String, String> map = new LinkedHashMap<>();

                for (Cell cell : result.listCells()) {
                    String family = Bytes.toString(CellUtil.cloneFamily(cell));
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    String row = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());

                    map.put("row", row);
                    map.put(family + ":" + qualifier, value);
                }
                serializedResults.add(map);
            }
            return serializedResults;
        }
    }


    public static void main(String[] args) throws IOException {
//        boolean atlas = isTableExist("atlas");
//        System.out.println(atlas);
//        List<String> cfs = new ArrayList<>();
//        cfs.add("id");
//        cfs.add("name");
//        System.out.println("createTable  " + createTable("sun_hong_yun", cfs));
////        System.out.println("putRow ) = " + putRow("sun_hong_yun", "2", "id", "id", "1"));
//        System.out.println("putRow ) = " + putRow("sun_hong_yun", "3", "name", "name", "1"));
//        System.out.println("selectRow() = " + selectRow("sun_hong_yun", "2"));
//
//        System.out.println(selectValue("sun_hong_yun", "2", "name", "id"));
        String taleName = "sun_test";

//        List<Map<String, Object>> allData = getAllData(taleName);
//        System.out.println(allData);

        List<Map<String, String>> maps = queryPage(taleName, 2, 2, "族1", "列1");
        String jsonString = JSON.toJSONString(maps);

        System.out.println(jsonString);

        PageResult pageResult = queryPage(taleName, "行1", 1, "族1", "列1");
        System.out.println(JSON.toJSONString(pageResult.getResultsAsMaps()));


    }
}